package com.saic.data.job;

import java.util.UUID;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.log4j.Logger;

import com.saic.data.dotask.DoTask;
import com.saic.data.util.FileUtil;

/**
 * @DESC:读锁
 * @author Tansheng
 */
public class ReadJob {

	private String projectName = "finance_platform";
	private String zookeeperAddress = "10.32.47.104:2181";
	private CuratorFramework client;
	private String jobID = UUID.randomUUID().toString();
	private static Logger logger = Logger.getLogger(ReadJob.class);
	private int retryTimes = 20;// 20 minutes
	private String readLockPath = "/locks/readLock";
	private String writeLockPath = "/locks/writeLock";
	private String readLockName = "readLock1122";

	public ReadJob() throws Exception {

		RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 5);
		this.client = CuratorFrameworkFactory.builder().connectString(this.zookeeperAddress).retryPolicy(retryPolicy)
				.namespace(this.projectName).build();
		client.start();
		try {
			retryTimes = Integer.parseInt(new String(client.getData().forPath("/conf/read_wait_minutes")));
		} catch (Exception e) {
		}
	}

	public void doReadJob(DoTask doTask) throws Exception {
		try {
			if (this.getReadLock()) {
				int waitMinutes = 0;
				do {
					waitMinutes++;
					if (waitMinutes >= this.retryTimes * 5) {
						throw new Exception("Wait write lock for: " + this.retryTimes + " mintues");
					}
					logger.info("wait for write lock..... [" + waitMinutes + "] "
							+ GetLock.isEmptyLock(this.client, this.writeLockPath));
				} while (!GetLock.isEmptyLock(this.client, this.writeLockPath).isEmpty());
				logger.info("Get read lock!");
				doTask.doTask();// TODO
			} else {
				logger.info("Read job already exists!");
			}
		} catch (Exception e) {
			throw e;
		} finally {}
	}

	private boolean getReadLock() throws Exception {
		return GetLock.getLock(this.client, this.jobID, this.readLockPath, FileUtil.getMD5(readLockName));
	}
	
	public void close(){
		if(null!=this.client)
			this.client.close();
	}

}
